package com.zh.test

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}


object SendKafkaMessage {
  val topic = "dev_test_hudi_1026"
  private val brokers = "dev-server001:9092,dev-server002:9092,dev-server003:9092"

  private val prop = new Properties()
  prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
  prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.StringSerializer")
  prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    "org.apache.kafka.common.serialization.StringSerializer")

  private val producer = new KafkaProducer[String, String](this.prop)

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("DakaMemProducer").setMaster("local[2]")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    val jsonDS =
      """
        |{
        |	"data": [{
        |		"id": "151105045279079999",
        |		"follow_fly_id": "1511050452772808898",
        |		"agent_number": "KY10287939501",
        |		"lading_number": "KY10287939501",
        |		"agent": "江东跨越",
        |		"rate": "0.500",
        |		"transport_mode": "20",
        |		"main_waybill_flag": "10",
        |		"agent_phone": "021-69730227",
        |		"flight_number": "浙A9T189",
        |		"flight_date": "2019-12-21 00:00:00",
        |		"flight_time": "02:00-13:20",
        |		"flight_takeoff_time": "2019-12-21 02:00:00",
        |		"flight_landing_time": "2019-12-21 13:20:00",
        |		"flight_status": "40",
        |		"miss_goods_flag": null,
        |		"miss_goods_reason": null,
        |		"miss_goods_handle": null,
        |		"freight_rate": "0.0",
        |		"freight_cost": "0.0",
        |		"change_flight_number": null,
        |		"change_flight_date": null,
        |		"change_flight_time": null,
        |		"change_flight_takeoff_time": null,
        |		"change_flight_landing_time": null,
        |		"change_freight_rate": "0.0",
        |		"starting_airport": "江东",
        |		"destination_airport": "鄂州",
        |		"agent_list_count": "193",
        |		"agent_list_weight": "2770.44",
        |		"agent_other_cost": "0.0",
        |		"agent_insurance_cost": "0.0",
        |		"agent_operate_rate": "0.0",
        |		"agent_operate_total_cost": "0.0",
        |		"agent_total_cost": "1411.6800",
        |		"out_agent_cost": "0.0",
        |		"out_agent_person": null,
        |		"agent_pay_mode": "10",
        |		"agent_sanction": "0.0",
        |		"agent_sanction_remark": null,
        |		"main_driver_id": null,
        |		"main_driver": "唐永禄",
        |		"main_driver_phone": "18868356653",
        |		"attach_driver_id": null,
        |		"attach_driver": null,
        |		"attach_driver_phone": null,
        |		"ticket_remark": null,
        |		"over_station_rate": "0.0",
        |		"over_station_cost": "0.0",
        |		"lading_date": "2019-12-21 00:00:00",
        |		"lading_by": null,
        |		"lading_by_name": "跨越速运集团有限公司",
        |		"lading_person_phone": null,
        |		"lading_cost": "0.0",
        |		"lading_other_cost": "0.0",
        |		"lading_rate": "0.0",
        |		"route_desc": "江东-鄂州-武汉",
        |		"trace_id": "d798489df4f602f04d58032453600e38",
        |		"enabled_flag": "1",
        |		"created_by": "68096",
        |		"creation_date": "2019-12-21 02:51:11",
        |		"updated_by": "System",
        |		"updation_date": "2019-12-24 16:08:27",
        |		"agent_id": "500000000000003100",
        |		"chartered_car_task_number": "RW191220-000323",
        |		"outer_transport_used_car_number": null,
        |		"agent_full_name": null,
        |		"agent_shipper": null,
        |		"agent_entruster": null,
        |		"lading_account_name": null,
        |		"lading_opening_bank": null,
        |		"lading_opening_account": null,
        |		"lading_belong_area": null,
        |		"lading_belong_point": null,
        |		"start_airport_code": null,
        |		"destination_airport_code": null,
        |		"miss_goods_audit": null,
        |		"miss_goods_remark": null,
        |		"arrivals_dispatch_id": "1000000917",
        |		"arrivals_dispatch": "鄂州赤湾二级中转场",
        |		"allowance_cost": "0.0"
        |	}],
        |	"database": "tms_followfly",
        |	"es": 1577174907000,
        |	"id": 34879410,
        |	"isDdl": false,
        |	"mysqlType": null,
        |	"old": [{
        |		"agent_total_cost": "1409.6800",
        |		"updation_date": "2019-12-21 14:04:37"
        |	}],
        |	"pkNames": null,
        |	"sql": null,
        |	"sqlType": null,
        |	"table": "follow_fly_agent",
        |	"ts": 1577174907421,
        |	"type": "UPDATE"
        |}
        |""".stripMargin

    import spark.implicits._
    val ds = spark.createDataset(Seq(jsonDS))
    val casWebDF: DataFrame = spark.read.json(ds)

    casWebDF.printSchema()

    casWebDF.createOrReplaceTempView("casWebDF")

    spark.sql("select data,data[0].agent from casWebDF").show()

    kafkaProducerSend(jsonDS)

  }

  def kafkaProducerSend(args: String) {
    if (args != null) {
      val message = new ProducerRecord[String, String](topic, null, args)
      producer.send(message)
    }
  }

}
